package hzmg.akkacloud.server.core.handler;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.cluster.pubsub.DistributedPubSub;
import akka.cluster.pubsub.DistributedPubSubMediator;
import akka.routing.SmallestMailboxPool;
import com.google.common.collect.Maps;
import com.hzmg.akkacloud.common.RemoteConstant;
import com.hzmg.akkacloud.common.exception.AkkaCloudException;
import com.hzmg.akkacloud.common.model.ServiceInstance;
import com.hzmg.akkacloud.common.request.WorkerHeartbeat;
import com.hzmg.akkacloud.common.response.AskResponse;
import com.hzmg.akkacloud.common.serialize.JsonUtils;
import hzmg.akkacloud.server.common.utils.SpringUtils;
import hzmg.akkacloud.server.core.handler.impl.WorkerRequestAkkaHandler;
import hzmg.akkacloud.server.core.handler.impl.WorkerRequestHttpHandler;
import hzmg.akkacloud.server.remote.transport.starter.AkkaStarter;
import hzmg.akkacloud.server.remote.transport.starter.VertXStarter;
import hzmg.akkacloud.server.remote.worker.WorkerClusterManagerService;
import hzmg.akkacloud.server.remote.worker.WorkerClusterQueryService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Map;

/**
 * receive and process worker's request
 *
 * @author zbw
 * @since 2022/7/17
 */
@Slf4j
@Component
public class WorkerRequestHandler {

    private static WorkerRequestHandler workerRequestHandler;
    @Resource
    private Environment environment;

    @Resource
    private WorkerClusterQueryService workerClusterQueryService;

    public static WorkerRequestHandler getWorkerRequestHandler() {
        if (workerRequestHandler == null) {
            workerRequestHandler = SpringUtils.getBean(WorkerRequestHandler.class);
        }
        return workerRequestHandler;
    }

    @PostConstruct
    public void initHandler() {
        // init server heartbeat handle actor
        ActorSystem actorSystem = AkkaStarter.actorSystem;
        actorSystem.actorOf(Props.create(WorkerRequestAkkaHandler.class)
                .withDispatcher("akka.server-actor-dispatcher")
                .withRouter(new SmallestMailboxPool(Runtime.getRuntime().availableProcessors() * 4)), RemoteConstant.SERVER_ACTOR_NAME);
        // init receive heartbeat from another registry.
        ActorRef clientHeartActor = actorSystem
                .actorOf(new SmallestMailboxPool(Runtime.getRuntime().availableProcessors() * 2)
                        .props(Props.create(ClientHeartReceiveActor.class)), RemoteConstant.SERVER_RECEIVE_HEARTBEAT_ACTOR_NAME);
        // subscribe
        ActorRef mediator = DistributedPubSub.get(actorSystem).mediator();
        // subscribe to the topic named "client_heart"
        mediator.tell(new DistributedPubSubMediator.Subscribe(RemoteConstant.HEARTBEAT_TOPIC, clientHeartActor), clientHeartActor);
        // init vert.x
        VertXStarter.vertx.deployVerticle(new WorkerRequestHttpHandler());
    }

    /**
     * 处理 Worker 的心跳请求
     * 同时返回最新service信息
     *
     * @param heartbeat 心跳包
     */
    public AskResponse onReceiveWorkerHeartbeat(WorkerHeartbeat heartbeat) throws AkkaCloudException {
        //test
        //log.info("receive the heartbeat from client:{}",heartbeat.getClient());
        WorkerClusterManagerService.updateStatus(heartbeat);
        Map<String, ServiceInstance> appName2ServiceInstance = Maps.newConcurrentMap();
        workerClusterQueryService.getAppId2ClusterStatus().forEach((c1, c2) -> {
            ServiceInstance serviceInstance = new ServiceInstance(c2.getAppName(), c2.getAddress2WorkerInfo());
            appName2ServiceInstance.computeIfAbsent(c2.getAppName(), ignore -> serviceInstance);
        });
        AskResponse askResponse = new AskResponse();
        if (!appName2ServiceInstance.isEmpty()) {
            askResponse.setSuccess(true);
            askResponse.setData(JsonUtils.toBytes(appName2ServiceInstance));
        } else {
            askResponse.setSuccess(false);
            askResponse.setMessage("this is a bug,please contact the author(@hzmg) to fix the bug!");
        }

        return askResponse;
    }

    /*    *//**
     * 处理OMS在线日志请求
     *
     * @param req 日志请求
     *//*
    public void onReceiveWorkerLogReportReq(WorkerLogReportReq req) {
        // 这个效率应该不会拉垮吧...也就是一些判断 + Map#get 吧...
        instanceLogService.submitLogs(req.getWorkerAddress(), req.getInstanceLogContents());
    }*/


}
